AWS AthenaでCloudTrailのS3オブジェクトログを解析をしてみました！

モバイルアプリサービス部の五十嵐です。

はじめてAthenaを使ってみたところクエリで苦戦して社内のメンバーに助けてもらったので、きっと他にも困っている人がいるだろうと思い記事にまとめました。

目的

これまでS3オブジェクトの操作ログは、S3のログ記録機能を使い、S3に出力されたログをLambdaからElasticsearch Serviceに登録し、Kibanaで参照していました。（以下は関連記事。）

• S3の操作ログをLambdaを経由してElasticsearchに送る方法 ｜ Developers.IO
• AmazonS3のサーバーアクセスログを解析 〜 最終版 ｜ Developers.IO

今回は、上記の仕組みを「CloudTrailのS3オブジェクトログ＋Athena」に置き換えが可能かどうかを検証してみました。

ちなみに、CloudTrailのS3オブジェクトログも最近追加された機能です。詳しくは以下を参照してください。

• 【新機能】CloudTrailでS3オブジェクトレベルのアクセスをロギングする ｜ Developers.IO

CloudTrailの設定

まずはCloudTrailの証跡情報の設定をします。

監視対象バケット（cm-athena-test）内のオブジェクトが操作（GET/PUT/DELETE/など）されたログが、ログ保管バケット（cm-athena-test-log）に保管されるように設定します。

• cm-athena-testバケットが監視対象のバケット
• cm-athena-test-logバケットがログ保管用のバケット

Athenaの設定

データベース定義

ウィザートに従い入力していきます。

Location of Input Data Setはデータベースのレコードとして扱うデータが置かれているS3のパスを設定します。 CloudTrailのログは s3://cm-athena-test-log/AWSLogs/XXXXXXXXXXXX/CloudTrail/ap-northeast-1/yyyy/mm/dd/*.json.gz というパスに保管されますので、固定値として設定できるまでのパス s3://cm-athena-test-log/AWSLogs/XXXXXXXXXXXX/CloudTrail/ap-northeast-1/ を設定します。

フォーマットはJSONです。ファイルはGZip圧縮されていますがAthenaはそのまま読み込んでくれます。

次にカラムを定義なのですが、CloudTrailのログフォーマットはArrayの中にJSONのログレコードが複数ある構造になっているため、ウィザードでは作成することはできませんでした。

ログファイルのデータ構造。

{
"Records": [
{
...
"eventTime": "2016-12-07T04:49:40Z",
"eventSource": "s3.amazonaws.com",
"eventName": "GetObject",
...
},
{
...
"eventTime": "2016-12-07T04:49:40Z",
"eventSource": "s3.amazonaws.com",
"eventName": "GetObject",
...
},
]
}

（このままウィザードを進めることでデータベースは作成されます。）

テーブル定義

ウィザードから定義ができないので諦めかけたのですが、社内のメンバーからHiveのSerDeを使っているっぽいので、その文法に従えばできるのではというアドバイスをもらいやってみたところ、以下のクエリでこのデータ構造のカラムを作ることができました。

CREATE EXTERNAL TABLE IF NOT EXISTS cm-athena-test.records (
records ARRAY<STRUCT<eventTime:STRING,eventSource:STRING,eventName:STRING>>
)
ROW FORMAT SERDE 'org.openx.data.jsonserde.JsonSerDe'
WITH SERDEPROPERTIES (
'serialization.format' = '1'
) LOCATION 's3://cm-athena-test-log/AWSLogs/XXXXXXXXXXXX/CloudTrail/ap-northeast-1/'

検索クエリ

作成したテーブルのカラムはArray型なのでそのまま検索すると配列が返ります。

SELECT * FROM records LIMIT 10;

配列の1要素ごとに対して条件指定をするには、 UNNEST 構文を利用することで絞り込むことができました。

SELECT record
FROM records
CROSS JOIN UNNEST(records) AS t (record)
WHERE record.eventName = 'PutObject';

まとめ

当初の目的であった「Athenaに置き換えが可能かどうか」については実現できることがわかりました。ただAthenaはKibanaほど検索方法が優しくはないので、フロントのWeb画面を用意してあげてクエリはJDBCで実行するようにすれば良さそうです。

他にもS3に蓄積されている大量のログの調査などに使えそうと思いました。ただデータ構造は配列ではなく改行区切りの方が扱いやすくて良いですね。もし自分でデータ構造を定義することがあれば検索することも考慮して設計しようと思いました。

参考

• HiveでJSONデータを処理するあれこれ(中級編) - Qiita
• Hadoop利用者ならきっと知ってる、Hive/Prestoクエリ関数の挙動の違い - Qiita